package com.zhl.hadoop.hbase;

import com.zhl.weibo.bean.KeyValBean;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/**
 * DDL:
 * 1.判断表是否存在
 * 2.创建表
 * 3.创建命名空间
 * 4.删除表
 *
 * DML：
 * 5.插入数据
 * 6.查数据
 * 7.查数据
 * 8.删除数据
 *
 * @author: 刘振华
 * @create: 2020-10-31 18:12
 **/
public class HbaseOperate {
	private Connection conn;
	private Admin ddlAdmin;
	Table table ;
	public static void main(String[] args) {
		HbaseOperate hbaseCli = new HbaseOperate();
//		hbaseCli.hbaseDdlTest();
		hbaseCli.hbaseDmlTest();
	}

	public void testDml()throws Exception{
		String tableName = "hive_test:visitor";
		String cf = "info";


//		String dKey = System.currentTimeMillis()+"";
//		System.out.println(dKey);
//		String column = "uid";
//		String value = "u01";

		String dKey = "1605172680897";

		String column = "vdate";
		String value = "2017/2/21";

//		String column = "vcount";
//		String value = "4";
//
		put(tableName,dKey,cf,column,value);

//		get(tableName, rowkey, cf, column);

		List<KeyValBean> rs = scan(tableName, cf, null, null, null, null);
		rs.forEach(System.out::println);
//
//		deleteData(tableName,rowkey,cf,column);
//
//		System.out.println("==============================");
//
//		scan(tableName,cf,column);


	}

	public List<KeyValBean> scan(String tableName, String cf, String column, Integer version, String startRow, String stopRow){
		Connection conn = null;
		Table table = null;
		Configuration conf = HBaseConfiguration.create();
		try {
			conn = ConnectionFactory.createConnection(conf);
			TableName tableObj = TableName.valueOf(tableName);
			table = conn.getTable(tableObj);
			Scan scan = new Scan();//scan可以传范围
			if(cf!=null && !"".equals(cf)){
				if(column!=null&&!"".equals(column)){
					scan.addColumn(Bytes.toBytes(cf),Bytes.toBytes(column));
				}else{
					scan.addFamily(Bytes.toBytes(cf));
				}
			}
			if(startRow!=null&&!"".equals(startRow)){
				scan.withStartRow(Bytes.toBytes(startRow));
			}

			if(stopRow!=null&&!"".equals(stopRow)){
				scan.withStopRow(Bytes.toBytes(stopRow));
			}
			if(version != null && version >0) {
				scan.readVersions(version);
			}
			final List<KeyValBean> rbList = new ArrayList<>();
			for(Result rs : table.getScanner(scan)){
				rs.listCells().forEach(c-> rbList.add(KeyValBean.builder()
					.rowKey(Bytes.toString(CellUtil.cloneRow(c)))
					.cf(Bytes.toString(CellUtil.cloneFamily(c)))
					.colName(Bytes.toString(CellUtil.cloneQualifier(c)))
					.value(Bytes.toString(CellUtil.cloneValue(c)))
					.timestamp(c.getTimestamp())
					.build()));
			}
			return rbList;
		} catch (IOException e) {
			e.printStackTrace();
		}finally {
			try {
				if(null != table) {
					table.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
			try {
				if(null != conn) {
					conn.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return null;
	}

	public void deleteData(String tableName, String rowKey, String cf, String column)throws Exception{
		table = conn.getTable(TableName.valueOf(tableName));
		Delete target = new Delete(Bytes.toBytes(rowKey));
//		target.addColumns(Bytes.toBytes(cf),Bytes.toBytes(column));
		target.addFamily(Bytes.toBytes(cf));
		table.delete(target);
	}

	public void scan(String tableName,String cf,String column)throws Exception{
		table = conn.getTable(TableName.valueOf(tableName));
		Scan scan = new Scan();//scan可以传范围
//		scan.withStartRow(Bytes.toBytes("abc111"));
//		scan.withStopRow(Bytes.toBytes("abc112"),true);
//		scan.addFamily(Bytes.toBytes(cf));
//		scan.addColumn(Bytes.toBytes(cf),Bytes.toBytes(column));
		ResultScanner scanner = table.getScanner(scan);
		for (Result it : scanner) {
			it.listCells().forEach(c-> System.out.println(Bytes.toString(CellUtil.cloneRow(c))+"\t"+
				Bytes.toString(CellUtil.cloneFamily(c))+"\t"+
				Bytes.toString(CellUtil.cloneQualifier(c))+"\t"+
				Bytes.toString(CellUtil.cloneValue(c))+"\t"+
				c.getTimestamp()));
		}
	}

	//获取数据
	public void get(String tableName, String rowKey, String cf, String column)throws Exception{
		table = conn.getTable(TableName.valueOf(tableName));
		Get get = new Get(Bytes.toBytes(rowKey));
//		get.addFamily(Bytes.toBytes(cf));
		get.addColumn(Bytes.toBytes(cf),Bytes.toBytes(column));
		get.readAllVersions();
		Result result = table.get(get);
		result.listCells().forEach(c-> System.out.println(Bytes.toString(CellUtil.cloneFamily(c))+","+
			Bytes.toString(CellUtil.cloneQualifier(c))+","+
			Bytes.toString(CellUtil.cloneValue(c))));
	}

	//向表中插入数据
	public void put(String tableName, String rowKey, String cf, String column, String value)throws Exception{
		table = conn.getTable(TableName.valueOf(tableName));
		Put p = new Put(Bytes.toBytes(rowKey));
		p.addColumn(Bytes.toBytes(cf), Bytes.toBytes(column), Bytes.toBytes(value));
		table.put(p);
	}

	public void hbaseDmlTest(){
		Configuration conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum","C111908191,C111908192,C111908193");
		try {
			conn = ConnectionFactory.createConnection(conf);
			testDml();

		}catch (Exception e){
			e.printStackTrace();
		}finally {
			try{
				if(null != table) {
					table.close();
				}
				if(null != conn) {
					conn.close();
				}
			}catch (Exception e){
				e.printStackTrace();
			}
		}
	}

	public void testDdl()throws Exception{
		//是否存在
//		isExist("myworld");

		String namespace = "hive_test";
		String tableName = namespace+":"+"visitor";
//		String tableName = "teacher_region";

		createTable(tableName,"info","other");

		//创建命名空间
//		createNamespace(namespace);
//		byte[][] bytes = {Bytes.toBytes("100000"),
//			Bytes.toBytes("200000"), Bytes.toBytes("400000"),
//			Bytes.toBytes("500000")};
//		//创建表
//		createTable(tableName,bytes,"basic","other");

		//删除表
//		deleteTable(tableName);

		//删除命名空间
//		dropNameSpace(namespace);
	}


	public void hbaseDdlTest(){
		Configuration conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum","C111908191,C111908192,C111908193");
		try {
			conn = ConnectionFactory.createConnection(conf);
			ddlAdmin = conn.getAdmin();

			testDdl();

		}catch (Exception e){
			e.printStackTrace();
		}finally {
			try{
				if(null != ddlAdmin) {
					ddlAdmin.close();
				}
				if(null != conn) {
					conn.close();
				}
			}catch (Exception e){
				e.printStackTrace();
			}
		}
	}

	//判断表是否存在
	public boolean isExist(String tableName)throws Exception{
		return ddlAdmin.tableExists(TableName.valueOf(tableName));
	}

	//创建表
	public void createTable(String tableName,String... cfs)throws Exception{
		if(isExist(tableName)){
			System.out.println(tableName+"表已经存在");
			return;
		}
		if(null == cfs || cfs.length < 1){
			System.out.println("列族不能为空！");
			return;
		}

		List<ColumnFamilyDescriptor> cfList = Arrays.asList(cfs).stream()
			.map(ColumnFamilyDescriptorBuilder::of)
			.collect(Collectors.toList());

		TableDescriptor tableDescriptor = TableDescriptorBuilder
			.newBuilder(TableName.valueOf(tableName))
			.setColumnFamilies(cfList)
			.build();


		ddlAdmin.createTable(tableDescriptor);
	}

	//创建表
	public void createTable(String tableName,byte[][] splitKeys,String... cfs)throws Exception{
		if(isExist(tableName)){
			System.out.println(tableName+"表已经存在");
			return;
		}
		if(null == cfs || cfs.length < 1){
			System.out.println("列族不能为空！");
			return;
		}

		List<ColumnFamilyDescriptor> cfList = Arrays.asList(cfs).stream()
			.map(ColumnFamilyDescriptorBuilder::of)
			.collect(Collectors.toList());

		TableDescriptor tableDescriptor = TableDescriptorBuilder
			.newBuilder(TableName.valueOf(tableName))
			.setColumnFamilies(cfList)
			.build();
		ddlAdmin.createTable(tableDescriptor,splitKeys);
	}

	//删除表
	public void deleteTable(String tableName)throws Exception{
		if(!isExist(tableName)){
			System.out.println(tableName+"表不存在!");
			return;
		}
		TableName tn = TableName.valueOf(tableName);
		ddlAdmin.disableTable(tn);
		ddlAdmin.deleteTable(tn);
	}

	//创建命名空间
	public void createNamespace(String namespace){
		try {
			ddlAdmin.createNamespace(NamespaceDescriptor.create(namespace).build());
		}catch (NamespaceExistException nse){
			nse.printStackTrace();
			System.out.println("命名空间已存在!");
		}catch (IOException e){
			e.printStackTrace();
		}
		System.out.println("存在了也能成功!");
	}

	//删除命名空间
	public void dropNameSpace(String namespace)throws Exception{
		ddlAdmin.deleteNamespace(namespace);
	}
}
